/*
 * Copyright 2008 Wink Saville
 * 
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License version 3
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

package com.saville.mc;

import java.lang.Thread;
import java.net.*;
import java.io.*;
import java.util.concurrent.atomic.*;

import com.saville.debug.*;
import com.saville.msgcomp.*;
import com.saville.serdes.*;

public class TcpMsgPipe {

    /*
     * Constructor
     * 
     * Open a MsgPipe using an existing socket
     */
    public TcpMsgPipe(Socket clientSocket, int producerDstId, int producerDstSubId) {
        mTcpMsgPipeName = "TcpMsgPipe-" + Timing.readTsc();
        Log.d(df, "%s: TcpMsgPipe(clientSocket, producerMc) E", mTcpMsgPipeName);
        init(clientSocket, producerDstId, producerDstSubId);
        Log.d(df, "%s: TcpMsgPipe(clientSocket, producerMc) X", mTcpMsgPipeName);
    }

    /*
     * Constructor
     * 
     * Open a MsgPipe using the url
     */
    public TcpMsgPipe(String clientUrl, int procuderDstId, int producerDstSubId) {
        if (!open(clientUrl, procuderDstId, producerDstSubId)) {
            throw new RuntimeException("TcpMsgPipe: Unable to open clientUrl");
        }
    }

    /*
     * Queue a message which will be sent down the pipe.
     */
    public void queueMsg(Msg msg) {
        mConsumer.queueMsg(msg);
    }

    /*
     * Open
     * 
     * TODO: Maybe make async and return a message when complete?
     */
    private boolean open(String clientUrl, int producerDstId, int producerDstSubId) {
        Socket clientSocket = null;
        boolean result;

        mDstId = producerDstId;
        mDstSubId = producerDstSubId;

        /* TODO: Maybe throw error rather than returning boolean? */
        try {
            URI uri = new URI(clientUrl);
            String scheme = uri.getScheme();
            int port = uri.getPort();
            if ((scheme != null) && (scheme.compareToIgnoreCase("amc") == 0)) {
                clientSocket = new Socket();
                if (port == -1) {
                    port = 46000;
                }
                Log.print("%s: connect to host=%s port=%d", mTcpMsgPipeName, uri.getHost(), port);
                clientSocket.connect(new InetSocketAddress(uri.getHost(), port));
                result = init(clientSocket, producerDstId, producerDstSubId);
            } else {
                result = false;
            }
        } catch (URISyntaxException uriE) {
            Log.print("%s: RISyntaxException e=%s", mTcpMsgPipeName, uriE);
            result = false;
        } catch (IOException e) {
            Log.e(df, "Error opening socket");
            result = false;
        }

        if (!result) {
            sendClosed();
        }

        return result;
    }

    /*
     * Initialize
     * 
     * TODO: Maybe make async and return a message when complete?
     */
    private boolean init(Socket clientSocket, int dstId, int dstSubId) {
        Log.d(df, "%s: init(clientSocket, producerMc) E", mTcpMsgPipeName);
        mMcMgr = McMgr.getInstance();

        mSocket = clientSocket;
        mDstId = dstId;
        mDstSubId = dstSubId;

        mProducer = new TcpPipeProducer();
        mConsumer = new TcpPipeConsumer();

        Log.d(df, "%s: init(clientSocket, producerMc) X", mTcpMsgPipeName);
        return true;
    }

    /*
     * Send one and only one CMD_CLOSED
     */
    private void sendClosed() {
        if (mOpened.getAndSet(false)) {
            Msg pipeMsg = new Msg();
            pipeMsg.guid = McConst.MCPIPE_GUID;
            pipeMsg.cmd = McConst.MCPIPE_CMD_CLOSED;
            pipeMsg.dstId = mDstId;
            pipeMsg.dstSubId = mDstSubId;
            mMcMgr.sendMsg(pipeMsg);
        }
    }

    /*
     * This class reads the pipe of messages sent by the corresponding
     * Consumer and produces messages which are sent to the mDstId with
     * mDstSubId.
     */
    private class TcpPipeProducer implements Runnable {
        public TcpPipeProducer() {
            mTcpPipeProducerName = "TcpPipeProducer" + mSocket.getLocalPort();
            Log.v(df, "%s: TcpPipeProducer(); E", mTcpPipeProducerName);

            Thread thr = new Thread(null, this, mTcpPipeProducerName);
            thr.start();

            Log.v(df, "%s: TcpPipeProducer(); X", mTcpPipeProducerName);
        }

        public void run() {
            byte data[] = new byte[16384];
            SerdesBinary serdes = new SerdesBinary();
            int MAX_LENGTH = 65536;

            try {
                Log.v(df, "%s: TcpPipeProducer; run() E", mTcpPipeProducerName);

                SocketAddress sa = mSocket.getRemoteSocketAddress();
                InetSocketAddress isa = (InetSocketAddress) sa;
                Log.v(df, "%s: remote inetAddress=%s, port=%s", mTcpPipeProducerName, isa
                        .getAddress().getHostAddress(), isa.getPort());

                mInStream = new DataInputStream(mSocket.getInputStream());

                while (true) {
                    int len = mInStream.readInt();
                    if (len > MAX_LENGTH) {
                        throw new Exception("Length too large");
                    }
                    if (len > data.length) {
                        data = new byte[len];
                    }
                    mInStream.readFully(data, 0, len);
                    serdes.init(data, 0, len);
                    Msg msg = (Msg) serdes.readObject();
                    Log.v(df, "%s: TcpPipeProducer; got msg guid=0x%x cmd=0x%x", mTcpPipeProducerName,
                        msg.guid, msg.cmd);

                    Msg pipeMsg = new Msg();
                    pipeMsg.guid = McConst.MCPIPE_GUID;
                    pipeMsg.cmd = McConst.MCPIPE_CMD_PRODUCER_MSG;
                    pipeMsg.dstId = mDstId;
                    pipeMsg.dstSubId = mDstSubId;
                    pipeMsg.obj = msg;

                    mMcMgr.sendMsg(pipeMsg);
                }
            } catch (Exception e) {
                Log.e(df, "%s: TcpPipeProducer; Exception e=%s", mTcpPipeProducerName, e);
                sendClosed();
            } finally {
                try {
                    if (mInStream != null)
                        mInStream.close();
                } catch (Exception e) {
                    Log.e(df, "%s: TcpPipeProducer; exception closing e=%s", mTcpPipeProducerName, e);
                }

                Log.v(df, "%s: TcpPipeProducer; run() X", mTcpPipeProducerName);
            }
        }

        private DataInputStream mInStream;
        private String mTcpPipeProducerName;
    }

    /*
     * This class consumes messages on its queue forwarding them to the producer
     * on the other side.
     */
    private class TcpPipeConsumer extends ActiveMc {
        public TcpPipeConsumer() {
            super("TcpPipeConsumer" + mSocket.getLocalPort());
            df = Log.VERBOSE;

            Log.v(df, "%s: TcpPipeConsumer(); E", mName);

            mSerdes = new SerdesBinary();

            try {
                OutputStream os = mSocket.getOutputStream();
                mOutStream = new DataOutputStream(os);
                // mOutStream = new DataOutputStream(mSocket.getOutputStream());
            } catch (Exception e) {
                Log.e(df, "%s: TcpPipeConsumer; Exception e=%s", mName, e);
                sendClosed();
            }

            Log.v(df, "%s: TcpPipeConsumer(); X", mName);
        }

        public void processMsg(Msg msg) {

            // Log.v(df, "%s: TcpPipeConsumer processMsg() E", mName);

            if (msg.is(McConst.MC_GUID, McConst.MC_CMD_STARTED)) {
                Log.v(df, "%s: TcpPipeConsumer; got STARTED", mName);
            } else {

                Log.v(df, "%s: TcpPipeConsumer; got msg guid=0x%x cmd=0x%x", mName, msg.guid, msg.cmd);

                try {
                    mSerdes.init();
                    msg.serialize(mSerdes);
                    byte data[] = mSerdes.data();
                    mOutStream.writeInt(mSerdes.size());
                    mOutStream.write(data, 0, mSerdes.size());

                    // retMsg(); // TODO: return the message
                } catch (Exception e) {
                    Log.e(df, "%s: TcpPipeConsumer exception processing e=%s", mName, e);
                    sendClosed();

                    try {
                        if (mOutStream != null)
                            mOutStream.close();
                    } catch (Exception ec) {
                        Log.e(df, "%s: TcpPipeConsumer exception closing ec=%s", mName, ec);
                    }
                }
            }

            // Log.v(df, "%s: TcpPipeConsumer processMsg() X", mName);
        }

        private SerdesBinary mSerdes;
        private DataOutputStream mOutStream;
    }

    private AtomicBoolean mOpened = new AtomicBoolean(true);
    private int mDstId;
    private int mDstSubId;
    private Socket mSocket;
    private String mTcpMsgPipeName;
    private TcpPipeProducer mProducer;
    private TcpPipeConsumer mConsumer;
    private int df = Log.VERBOSE;
    private IMcMgr mMcMgr;
}
